package com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.scheduling;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.controller.CommonResponse;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.utils.KafkaService;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.controller.ProduceSuccessResponse;
import com.qupeng.demo.kafka.kafkaspringbootstreams.demo.CustomerComplaint;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentCase;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentProfile;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.concurrent.ExecutionException;

//@Slf4j
//@Component
//@EnableScheduling
public class ScheduleTask {
    @Autowired
    private KafkaService<String, AgentProfile> profileService;

    @Autowired
    private KafkaService<String, AgentCase> caseService;

    @Autowired
    private KafkaService<String, CustomerComplaint> complaintService;

    @Scheduled(fixedRate = 10000)
    public void scheduleTask() throws ExecutionException, InterruptedException {
        profileService.sendToKafkaSync("agent-profile", "u001", new AgentProfile("u001", "Peter", 5));
//        caseService.sendToKafkaSync("agent-case", "u001", new AgentCase("c001", AgentCase.Status.COMPLETED));
//        complaintService.sendToKafkaSync("agent-case", "u001", new );
    }

    private <K, V> CommonResponse<ProduceSuccessResponse<K, V>> produce(KafkaService<K, V> kafkaService, V messagePayload, String topic, K key) {
        try {
            SendResult<K, V> sendResult = kafkaService.sendToKafkaSync(topic, key, messagePayload);
            return new CommonResponse<>("producer", new ProduceSuccessResponse<K, V>(sendResult));
        } catch (ExecutionException | InterruptedException e) {
            return new CommonResponse<>("001-producer-controller", e.getMessage(), "producer");
        }
    }
}